package com.noble.glamour.module.logic.socket.stream;

import com.noble.glamour.module.logic.socket.NetData;
import com.noble.glamour.module.logic.socket.listener.SocketConnectionListener;
import com.noble.glamour.module.logic.socket.logic.KeepAliveSocket;
import com.noble.glamour.module.logic.socket.logic.task.SocketIoHandler;
import com.noble.glamour.module.logic.socket.logic.task.SubTread;
import com.noble.glamour.module.logic.socket.logic.task.ThreadRunner;
import com.noble.glamour.module.logic.socket.utils.IMUtils;
import com.noble.glamour.module.logic.socket.utils.SocketState;
import com.noble.library.log.PLogger;

import java.io.IOException;
import java.io.OutputStream;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

/**
 * 写入数据包
 */
public class PacketWriter implements ThreadRunner {
    private final int SOCKET_PACKET_QUEUE_SIZE = 500;

    private final ArrayBlockingQueue<NetData> queue = new ArrayBlockingQueue(SOCKET_PACKET_QUEUE_SIZE, true);
    private Executor singleSendErrorExecutor = Executors.newSingleThreadExecutor();
    private volatile boolean endWithException = false;
    private boolean isShutDown = false;
    private SubTread writerThread;

    private SocketConnectionListener listener;    // socket连接状态监听
    private SocketState state;

    private OutputStream output;   // 输出流

    public PacketWriter(SocketState state, SocketConnectionListener listener) {
        this.state = state;
        this.listener = listener;
    }

    @Override
    public void run() {
        writePacket();
    }

    public void init(KeepAliveSocket client, OutputStream output) {
        endWithException = false;

        this.output = output;
        writerThread = new SubTread(client, this);
        writerThread.start();
        IMUtils.log("Socket send packet thread init and start");
    }

    public void sendPacket(final NetData data) {
        try {
            queue.put(data);
        } catch (Exception e) {
            if (listener != null) {
                singleSendErrorExecutor.execute(() -> listener.onSendPacketError(state, data));
            }
            IMUtils.log("Socket send packet add to queue error:" + e.getMessage());
        }
    }

    private void writePacket() {
        NetData packet = null;
        //loop
        try {
            while (!done()) {
                packet = queue.take();
                if (packet == null) {
                    Thread.sleep(100);
                    continue;
                }
                IMUtils.log("Socket send packet :" + packet.toString());

                byte[] data = packet.getBytes();
                output.write(data);
                output.flush();
                packet = null;
            }
        } catch (InterruptedException e) {
            IMUtils.log("Socket send packet ,take packet from queue interrupt");
        } catch (IOException e) {
            final NetData errorPacket = packet;
            if (listener != null) {
                singleSendErrorExecutor.execute(() -> listener.onSendPacketError(state, errorPacket));
            }
            endWithException = true;
        } catch (Exception e) {
            PLogger.printThrowable(e);
        }

        //fail left msg
        do {
            packet = queue.poll();
            if (packet == null) {
                break;
            }
            final NetData errorPacket = packet;
            if (listener != null) {
                singleSendErrorExecutor.execute(() -> listener.onSendPacketError(state, errorPacket));
            }
        } while (true);

        //注意消息顺序
        if (endWithException) {
            writerThread.handler.sendEmptyMessage(SocketIoHandler.IO_ERROR);
        }
        writerThread.handler.sendEmptyMessage(SocketIoHandler.WRITER_CLOSE);

        IMUtils.log("Socket send packet thread shutdown done ,instant with exception:" + endWithException);
    }

    public boolean isEndWithException() {
        return endWithException;
    }

    private boolean done() {
        return isShutDown;
    }

    public void shutdown(boolean instant) {
        if (isShutDown) {
            return;
        }
        IMUtils.log("Socket send packet thread shutdown instant:" + instant + " start");
        isShutDown = true;
        writerThread.interrupt();
        IMUtils.log("Socket send packet thread shutdown instant:" + instant + " end");
    }
}
